package timeandwindow;

/*
*
* 增量聚合函数
*   AggregateFunction: 两两聚合，和reduce相比使用范围更广，支持输入和输出类型不一致。
*       三个泛型：
*          <IN> -The type of the values that are aggregated(input values) 输入数据类型
*          <ACC> - The type of the accumulator (intermediate aggregate state) 累加器类型
*          <OUT> – The type of the aggregated result 输出数据类型
*  四个方法：
*      createAccumulater():创建累加器，只会调用一次
*      add():进行累加操作，每条数据调用一次
*      getResult():获取结果，只会调用一次
*      merge():合并累加器
* */

/*
* 思路：
*   1.确定窗口
* */

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import util.SourceUtil;

import java.time.Duration;
import java.util.HashSet;

public class Flink07_AggregateFunction {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);
         SingleOutputStreamOperator<Event> ds = env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource")
                 .assignTimestampsAndWatermarks(
                         WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                 .withTimestampAssigner(
                                         (event, ts) -> event.getTs()
                                 )
                 );

         ds.print("INPUT");

         //需求：统计在10秒内的UV(独立访客数)
         //todo PV统计的是所有的点击量；而对用户id进行去重之后，得到的就是UV
         //todo 窗口：非按键分区时间滚动窗口：windowall,时间滚动函数忘了

         ds.windowAll(
                 TumblingEventTimeWindows.of(Time.seconds(10))
         ).aggregate(
                 //使用hashSet维护所有访问的人，hashset有天然去重功能，最后只要取hashset的长度即可表示UV

                 /**
                  * 三个泛型，in，acc，out
                  */
                 new AggregateFunction<Event, HashSet<String>, Long>() {

                     //四个方法：1.创建累加器
                     @Override
                     public HashSet createAccumulator() {
                         return new HashSet<>();//初始化一个空的HashSet，用于存储唯一的用户ID。
                     }


                     //2.进行累加操作，每条数据调用一次
                     @Override
                     public HashSet add(Event value, HashSet accumulator) {
                         System.out.println("add.......");
                         accumulator.add(value.getUser());//取一个一个对象名字添加到累加器hashset中保存，是集合可以去重
                         return accumulator;
                     }

                    //3.获取结果，只会调用一次
                     @Override
                     public Long getResult(HashSet accumulator) {
                         return (long)accumulator.size();
                     }

                    //4.合并累加器
                       @Override
                     public HashSet<String> merge(HashSet<String> a, HashSet<String> b) {
                         return null; //在分布式环境中，多个任务实例可能产生多个HashSet，这里的方法应该合并这些HashSet
                     }

                 }
         ).print("UV");


         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }


}
